package com.db.report.client;

import com.db.aggregation.AbstractAggregation;
import com.db.aggregation.ExchangeAggregation;
import com.db.aggregation.NullAggregation;
import com.db.aggregation.StockAndSideAggregation;
import com.db.protocol.message.ReportMessage;
import com.db.report.Storage;
import com.db.transport.client.SessionClient;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Logger;

/**
 * @author nulls.narik
 */
public class SenderClient {

    private static Logger logger = Logger.getLogger(SenderClient.class.getName());

    private final String clientID;
    private final SessionClient sessionClient;
    private final AbstractAggregation aggregation;
    private final Long timer;
    private final BlockingQueue<ReportMessage> queue;
    private final Storage storage;
    private Thread threadGetFromQueue = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                while (true) {
                    ReportMessage reportMessage = queue.take();
                    storage.add(reportMessage);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }

        }
    });
    private Thread threadSendMessages = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                while (true) {
                    Long time = System.currentTimeMillis();
                    sendMessages();
                    time = timer - (System.currentTimeMillis() - time);
                    if (time > 0) {
                        Thread.sleep(time);
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    });


    public SenderClient(String clientID, SessionClient sessionClient, Integer typeAggregation,
                        BlockingQueue<ReportMessage> queue, Long timer) {
        this.clientID = clientID;
        this.timer = timer;
        this.storage = new Storage();
        this.queue = queue;
        this.sessionClient = sessionClient;
        switch (typeAggregation) {
            case 1:
                aggregation = new ExchangeAggregation();
                break;
            case 2:
                aggregation = new StockAndSideAggregation();
                break;
            default:
                aggregation = new NullAggregation();
                break;
        }
    }

    public void startGetFromQueue() {
        threadGetFromQueue.start();
        logger.info(String.format("Client [%s] start thread for new message from queue.", clientID));
    }

    public void startSendMessages() {
        threadSendMessages.start();
        logger.info(String.format("Client [%s] start thread for send messages.", clientID));
    }

    protected void sendMessages() {
        logger.info(String.format("Client [%s] sending aggregating message.", clientID));

        List<ReportMessage> messages = storage.getAll();
        aggregation.aggregate(messages);
        messages = aggregation.getAll();
        for (ReportMessage message : messages) {
            sessionClient.sendMessage(message.toString());
        }
    }

}
